package org.example.kafka;

import cn.hutool.core.date.DateUtil;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;

import java.util.*;
import java.util.stream.Collectors;


public class KafkaConsumerCustomExample {

    public static final Logger log = LoggerFactory.getLogger("consumer");

    public static void main(String[] args) throws Exception {
        String topicName = "test4";

        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.236.128:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "app_hp1");
        props.put("client.id", "client_01");
        props.put("enable.auto.commit", true);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        long startTime = DateUtil.parseDateTime("2025-04-13 12:12:00").getTime();

        List<PartitionInfo> partitions = consumer.partitionsFor(topicName);
        List<TopicPartition> topicPartitionList = partitions
                .stream()
                .map(info -> new TopicPartition(topicName, info.partition()))
                .collect(Collectors.toList());
        consumer.assign(topicPartitionList);

        Map<TopicPartition, Long> partitionTimestampMap = topicPartitionList.stream()
                .collect(Collectors.toMap(tp -> tp, tp -> startTime));
        Map<TopicPartition, OffsetAndTimestamp> partitionOffsetMap = consumer.offsetsForTimes(partitionTimestampMap);
        // Force the consumer to seek for those offsets
        partitionOffsetMap.forEach((tp, offsetAndTimestamp) -> consumer.seek(tp, offsetAndTimestamp.offset()));

        boolean keepOnReading = true;
        while (keepOnReading) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                log.info("Message received {}, partition {} , offset={}, timestamp={}, dateStr={}", record.value(), record.partition(), record.offset(), record.timestamp(), DateUtil.date(record.timestamp()));
            }
        }
    }

}
